package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.DimAsyncFunction;
import com.atguigu.bean.ExamPaper;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import com.atguigu.utils.WindowFunctionUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * ClassName: Dws152_ExamCourse
 * Package: com.atguigu.app.dws
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/5/18 14:45
 * @Version 1.0
 */

//数据流: web/app -> Maxwell -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> ClickHouse(DWS)
//程 序:  Mock -> Maxwell -> Kafka(ZK) -> BaseDBApp -> Kafka(ZK) -> Dws152_ExamCourseWindow -> ClickHouse(ZK)
public class Dws152_ExamCourseWindow {

    public static void main(String[] args) throws Exception {

        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(18*60+5));
        //  状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //2.从kafka DWD层读取dwd_study_test_exam主题的数据
        DataStreamSource<String> kafkaDS = env.fromSource(MyKafkaUtil.getKafkaSource("dwd_test_exam", "exam_paper"),
                WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                    @Override
                    public long extractTimestamp(String element, long recordTimestamp) {
                        JSONObject jsonObject = JSON.parseObject(element);
                        return jsonObject.getLong("create_time");
                    }
                }),
                "kafka-source");

        //3.转换为JavaBean
        SingleOutputStreamOperator<ExamPaper> examPaperDS = kafkaDS.map(new MapFunction<String, ExamPaper>() {
            @Override
            public ExamPaper map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                return new ExamPaper("", "",jsonObject.getString("paper_id"),"","","", 1L, jsonObject.getBigDecimal(
                        "score"), jsonObject.getLong("duration_sec"),  0L);
            }
        });

        //4.维度关联获取courseId
        SingleOutputStreamOperator<ExamPaper> courseIdDS = AsyncDataStream.unorderedWait(examPaperDS,
                new DimAsyncFunction<ExamPaper>("DIM_TEST_PAPER") {
            @Override
            public String getkey(ExamPaper input) throws Exception {
                return input.getPaperId();
            }

            @Override
            public void join(ExamPaper input, JSONObject dimInfo) throws Exception {
                input.setCourseId(dimInfo.getString("COURSE_ID"));
                input.setPaperTitle(dimInfo.getString("PAPER_TITLE"));
            }
        }, 60, TimeUnit.SECONDS);


        //5.按照 course_id 进行Keyby 分组 开窗 聚合
        SingleOutputStreamOperator<ExamPaper> reduceDS =
                courseIdDS.keyBy(ExamPaper::getCourseId).window(TumblingEventTimeWindows.of(Time.seconds(10))).reduce(new ReduceFunction<ExamPaper>() {
            @Override
            public ExamPaper reduce(ExamPaper value1, ExamPaper value2) throws Exception {
                value1.setUserNum(value1.getUserNum() + value2.getUserNum());
                value1.setScore(value1.getScore().add(value2.getScore()));
                value1.setDurationSec(value1.getDurationSec() + value2.getDurationSec());

                return value1;
            }
        }, new WindowFunction<ExamPaper, ExamPaper, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<ExamPaper> input, Collector<ExamPaper> out) throws Exception {

                WindowFunctionUtil.set(input, window, out);
            }
        });

        reduceDS.print("reduceDS-->");
        //关联维表
        SingleOutputStreamOperator<ExamPaper> resultDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<ExamPaper>(
                "DIM_COURSE_INFO") {
            @Override
            public String getkey(ExamPaper input) throws Exception {
                return input.getCourseId();
            }

            @Override
            public void join(ExamPaper examPaper, JSONObject dimInfo) throws Exception {
                examPaper.setCourseName(dimInfo.getString("COURSE_NAME"));
            }

        }, 60, TimeUnit.SECONDS);


        //输出
        resultDS.print("resultDS-->");
        reduceDS.addSink(MyClickHouseUtil.getSinkFunction("insert into exam_course_window values(?,?,?,?,?,?,?,?,?,?)"));

        //执行
        env.execute("152-->");

    }
}
